package com.spark.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * Created by TRS on 2017/6/26.
  */
object pagerank {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("wordCount").setMaster("local")
      .set("spark.driver.memory", "2G")
    //创建Spark上下
    val sc = new SparkContext(conf)

    val path = "E:\\pagerank.txt"
    val lines = sc.textFile(path)
    val links = lines.map { s =>
      val parts = s.split("\\s+")
      (parts(0), parts(1))
    }.distinct().groupByKey().cache()

    links.foreach(x => println(x))


    var ranks = links.mapValues(v => 1.0)

    ranks.foreach(x => println(x))

    val iters = 5

    for (i <- 1 to iters) {

      links.join(ranks).foreach(x=>println(x))
      ranks.foreach(tup => println(tup))

      val contribs = links.join(ranks).values.flatMap { case (urls, rank) =>
        val size = urls.size
        urls.map(url => (url, rank / size))
      }
      ranks = contribs.reduceByKey(_ + _).mapValues(0.15 + 0.85 * _)


    }

    val output = ranks.collect().sortBy(x=>x._2)
    output.foreach(tup => println(tup._1 + " has rank: " + tup._2 + "."))


    sc.stop()
  }

}
